package com.hunliji.monitor.web.mq.rocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.hunliji.common.mq.MsgConverter;
import com.hunliji.common.mq.MsgData;
import com.hunliji.monitor.share.component.mq.rocket.constants.RocketMqConstants;
import com.hunliji.monitor.share.component.mq.rocket.impl.RocketMqFacadeImpl;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 测试监听
 * @author wuchen
 */
@Component
@RocketMQMessageListener(
        //topic配置
        topic = RocketMqConstants.TEST_TOPIC,
        //tag配置
        selectorExpression = RocketMqConstants.TEST_TAG,
        //消费者组配置
        consumerGroup = RocketMqConstants.TEST_CONSUME_GROUP
)
public class TestListener implements RocketMQListener<MessageExt> , RocketMQPushConsumerLifecycleListener {
    private static final Logger logger = LoggerFactory.getLogger(RocketMqFacadeImpl.class);

    /**
     * 消息的消费
     * @param message
     */
    @Override
    public void onMessage(MessageExt message) {
        try {
            MsgData<String> msgData = MsgConverter.getMsgData(message.getBody(), new TypeReference<MsgData<String>>(){});
            SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
            logger.info("business 测试监听消息 time:{},msgData:{}",sdf.format(new Date()), JSON.toJSONString(msgData));
        }catch (Exception e){
            logger.error("business 消息处理异常",e);
        }

    }

    /**
     * 可以对一些基本参数进行配置
     * @param consumer
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        //设置最大的重试次数，默认根据TCP协议的重试次数进行重试，模式16次
        consumer.setMaxReconsumeTimes(-1);
        //设置最大处理线程，这个是默认值，请根据实际情况进行配置
        consumer.setConsumeThreadMax(64);
        //设置最小处理线程
        consumer.setConsumeThreadMin(2);
        //配置消费方式，目前支持广播和集群的消费方式
        consumer.setMessageModel(MessageModel.CLUSTERING);
    }
}
